"""
设备管理 CRUD 操作 - 支持地址绑定
统一异常、日志、校验、分页风格
"""
from datetime import datetime
from typing import List, Optional, Tuple, Dict, Any

from app.models.device_models import Station, Box, RTU, Meter
from app.schemas.device_schemas import (
	DeviceQuery,
	StationCreate, StationUpdate,
	BoxCreate, BoxUpdate,
	RTUCreate, RTUUpdate,
	MeterCreate, MeterUpdate
)
from app.service.address_service import AddressService
from common.base_enums import DeviceStatus, DeviceType
from common.database import DatabaseService
from common.exceptions import BizException, ErrorCode
from common.logger import create_logger

logger = create_logger("device_crud")


def get_region_prefix(query: DeviceQuery) -> Optional[str]:
	"""获取区域前缀，用于区域过滤，优先使用area_code，其次是city_code，最后是province_code"""
	# 优先使用area_code
	if query.area_code:
		return query.area_code
	# 其次使用city_code
	elif query.city_code:
		return query.city_code[:4]  # 市编码通常为6位，取前4位作为前缀
	# 最后使用province_code
	elif query.province_code:
		return query.province_code[:2]  # 省编码通常为6位，取前2位作为前缀
	return None


# ---------- 台区操作 ----------

async def get_station_by_id(db: DatabaseService, station_id: int) -> Optional[Station]:
	sql = """
        SELECT id, code, serial_no, name, capacity_kva, manager_code, status,
               address_id, created_at, updated_at
        FROM dev_station
        WHERE id = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (station_id,))
		if not row:
			return None
		row["id"] = int(row["id"])
		return Station(**row)
	except Exception as e:
		logger.exception("查询台区失败，station_id=%s", station_id)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询台区失败") from e


async def get_station_with_address(db: DatabaseService, station_id: int) -> Optional[Dict[str, Any]]:
	"""获取台区及其地址信息"""
	station = await get_station_by_id(db, station_id)
	if not station:
		return None

	# 获取地址信息
	address_sql = """
        SELECT da.address_type, da.is_primary, da.description,
               ba.id as address_id, ba.name as address_name, ba.full_name,
               ba.longitude, ba.latitude
        FROM dev_device_address da
        INNER JOIN base_region_address ba ON da.address_id = ba.id
        WHERE da.device_type = 'STATION' AND da.device_id = %s AND da.deleted = 0 AND ba.is_active = 1
        ORDER BY da.is_primary DESC, da.created_at ASC
    """
	try:
		address_results = db.get_all(address_sql, (station_id,))
		location_info = []
		for addr in address_results:
			location_info.append({
				"address_id": addr["address_id"],
				"address_name": addr["address_name"],
				"full_address": addr["full_name"],
				"longitude": addr["longitude"],
				"latitude": addr["latitude"],
				"address_type": addr["address_type"],
				"is_primary": bool(addr["is_primary"])
			})

		station_dict = station.model_dump()
		station_dict["location_info"] = location_info
		return station_dict
	except Exception as e:
		logger.exception("查询台区地址信息失败，station_id=%s", station_id)
		# 如果地址查询失败，仍返回台区基本信息
		station_dict = station.model_dump()
		station_dict["location_info"] = []
		return station_dict


async def get_station_by_code(db: DatabaseService, code: str) -> Optional[Station]:
	sql = """
        SELECT id, code, serial_no, name, capacity_kva, manager_code, status,
               address_id, created_at, updated_at
        FROM dev_station
        WHERE code = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (code,))
		if not row:
			return None
		row["id"] = int(row["id"])
		return Station(**row)
	except Exception as e:
		logger.exception("查询台区失败，code=%s", code)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询台区失败") from e


async def get_station_by_serial_no(db: DatabaseService, serial_no: str) -> Optional[Station]:
	"""根据序列号查询台区"""
	sql = """
        SELECT id, code, serial_no, name, capacity_kva, manager_code, status,
               address_id, created_at, updated_at
        FROM dev_station
        WHERE serial_no = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (serial_no,))
		if not row:
			return None
		row["id"] = int(row["id"])
		return Station(**row)
	except Exception as e:
		logger.exception("查询台区失败，serial_no=%s", serial_no)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询台区失败") from e


async def get_station_list(db: DatabaseService, query: DeviceQuery) -> Tuple[int, List[Station]]:
	"""分页查询台区列表"""
	where = ["deleted = 0"]
	params: List[Any] = []

	if query.keyword:
		where.append("(code LIKE %s OR name LIKE %s OR serial_no LIKE %s)")
		params.extend([f"%{query.keyword}%", f"%{query.keyword}%", f"%{query.keyword}%"])
	if query.status:
		where.append("status = %s")
		params.append(query.status.value)
	if query.manager_code is not None:
		where.append("manager_code = %s")
		params.append(query.manager_code)
	if query.station_id:
		where.append("id = %s")
		params.append(query.station_id)

	where_clause = " AND ".join(where)
	count_sql = f"SELECT COUNT(*) AS count FROM dev_station WHERE {where_clause}"

	try:
		count_row = db.get_one(count_sql, params)
		total = count_row["count"] if count_row else 0
		if total == 0:
			return 0, []

		offset = (query.page - 1) * query.size
		select_sql = f"""
            SELECT id, code, serial_no, name, capacity_kva, manager_code, status,
                   address_id, created_at, updated_at
            FROM dev_station
            WHERE {where_clause}
            ORDER BY created_at DESC
            LIMIT %s OFFSET %s
        """
		params.extend([query.size, offset])
		rows = db.get_all(select_sql, params)

		stations = []
		for row in rows:
			row["id"] = int(row["id"])
			stations.append(Station(**row))
		return total, stations
	except Exception as e:
		logger.exception("分页查询台区失败，query=%s", query)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询台区列表失败") from e


async def create_station(db: DatabaseService, payload: StationCreate) -> int:
	"""
	新建台区接口：入参为基本信息、address_id和坐标值
	后端新增dev_station记录，然后通过AddressBindingService.bind写dev_device_address
	"""
	if await get_station_by_code(db, payload.code):
		logger.warning("创建台区失败，编码已存在：%s", payload.code)
		raise BizException.from_error_code(ErrorCode.CONFLICT, message=f"台区编码 {payload.code} 已存在")

	sql = """
        INSERT INTO dev_station (code, serial_no, name, capacity_kva, manager_code, status, address_id)
        VALUES (%s, %s, %s, %s, %s, %s, %s)
    """
	try:
		station_id = db.insert(sql, (
			payload.code,
			payload.serial_no,
			payload.name,
			payload.capacity_kva,
			payload.manager_code,
			payload.status.value if payload.status else DeviceStatus.ONLINE.value,
			payload.address_id
		))

		# 绑定地址
		await AddressService.bind(db, DeviceType.STATION.value, station_id, payload.serial_no, payload.address_id)

		logger.info("台区创建成功，station_id=%s, address_id=%s", station_id, payload.address_id)
		return station_id
	except Exception as e:
		logger.exception("创建台区失败，payload=%s", payload)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="创建台区失败") from e


async def update_station(db: DatabaseService, station_id: int, payload: StationUpdate) -> bool:
	"""
	更新台区接口：
	dev_device_address 增加 deleted、created_at、ended_at。更新时不update原记录，
	而是新增一条新address映射+结束旧记录。
	修改基本信息和address_id（可选），地址信息通过"系统管理"修改。
	"""
	station = await get_station_by_id(db, station_id)
	if not station:
		return False

	fields, params = [], []
	if payload.name is not None:
		fields.append("name = %s")
		params.append(payload.name)
	if payload.capacity_kva is not None:
		fields.append("capacity_kva = %s")
		params.append(payload.capacity_kva)
	if payload.manager_code is not None:
		fields.append("manager_code = %s")
		params.append(payload.manager_code)
	if payload.status is not None:
		fields.append("status = %s")
		params.append(payload.status.value)

	# 处理地址变更
	if hasattr(payload, 'address_id') and payload.address_id is not None and payload.address_id != station.address_id:
		await AddressService.change_address(db, DeviceType.STATION.value, station_id, station.serial_no,
											payload.address_id)
		fields.append("address_id = %s")
		params.append(payload.address_id)
		logger.info("台区地址变更，station_id=%s, old_address_id=%s, new_address_id=%s",
					station_id, station.address_id, payload.address_id)

	if not fields:
		return True  # 无更新
	params.append(station_id)

	sql = f"UPDATE dev_station SET {', '.join(fields)}, updated_at = CURRENT_TIMESTAMP WHERE id = %s AND deleted = 0"
	try:
		return db.update(sql, params) > 0
	except Exception as e:
		logger.exception("更新台区失败，station_id=%s, payload=%s", station_id, payload)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="更新台区失败") from e


async def delete_station(db: DatabaseService, station_id: int) -> bool:
	if not await get_station_by_id(db, station_id):
		return False

	try:
		async with db.transaction():
			# 逻辑删除设备地址记录
			delete_address_sql = """
                UPDATE dev_device_address
                SET deleted = 1, ended_at = CURRENT_TIMESTAMP
                WHERE device_type = 'STATION' AND device_id = %s
            """
			db.update(delete_address_sql, (station_id,))
			logger.info("已逻辑删除台区 %s 的设备地址记录", station_id)

			# 逻辑删除台区
			sql = """
                UPDATE dev_station
                SET deleted = 1, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
            """
			result = db.update(sql, (station_id,))

			logger.info("成功删除台区，station_id=%s", station_id)
			return result > 0
	except Exception as e:
		logger.exception("删除台区失败，station_id=%s", station_id)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="删除台区失败") from e


# ---------- 电箱操作 ----------

async def get_box_by_id(db: DatabaseService, box_id: int) -> Optional[Box]:
	sql = """
        SELECT d.id, d.station_id, d.code, d.serial_no, ds.serial_no as station_no, d.install_date, d.status, d.manager_code,
       			d.address_id, d.created_at, d.updated_at
		FROM dev_box d
		INNER JOIN dev_station ds ON d.station_id = ds.id
		WHERE d.id = %s AND d.deleted = 0
    """
	try:
		row = db.get_one(sql, (box_id,))
		if not row:
			return None
		return Box(**row)
	except Exception as e:
		logger.exception("查询电箱失败，box_id=%s", box_id)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电箱失败") from e


async def get_box_by_code(db: DatabaseService, code: str) -> Optional[Box]:
	sql = """
        SELECT id, station_id, code, serial_no, install_date, status, manager_code,
               address_id, created_at, updated_at
        FROM dev_box
        WHERE code = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (code,))
		if not row:
			return None
		row["id"] = int(row["id"])
		row["station_id"] = int(row["station_id"])
		return Box(**row)
	except Exception as e:
		logger.exception("查询电箱失败，code=%s", code)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电箱失败") from e


async def get_box_by_serial_no(db: DatabaseService, serial_no: str) -> Optional[Box]:
	"""根据序列号查询电箱"""
	sql = """
        SELECT id, station_id, code, serial_no, install_date, status, manager_code,
               address_id, created_at, updated_at
        FROM dev_box
        WHERE serial_no = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (serial_no,))
		if not row:
			return None
		row["id"] = int(row["id"])
		row["station_id"] = int(row["station_id"])
		return Box(**row)
	except Exception as e:
		logger.exception("查询电箱失败，serial_no=%s", serial_no)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电箱失败") from e


async def get_box_with_address(db: DatabaseService, box_id: int) -> Optional[Dict[str, Any]]:
	"""获取电箱及其地址信息"""
	box = await get_box_by_id(db, box_id)
	if not box:
		return None

	# 获取地址信息
	address_sql = """
        SELECT da.address_type, da.is_primary, da.description,
               ba.id as address_id, ba.name as address_name, ba.full_name,
               ba.longitude, ba.latitude
        FROM dev_device_address da
        INNER JOIN base_region_address ba ON da.address_id = ba.id
        WHERE da.device_type = 'BOX' AND da.device_id = %s AND da.deleted = 0 AND ba.is_active = 1
        ORDER BY da.is_primary DESC, da.created_at ASC
    """
	try:
		address_results = db.get_all(address_sql, (box_id,))
		location_info = []
		for addr in address_results:
			location_info.append({
				"address_id": addr["address_id"],
				"address_name": addr["address_name"],
				"full_address": addr["full_name"],
				"longitude": addr["longitude"],
				"latitude": addr["latitude"],
				"address_type": addr["address_type"],
				"is_primary": bool(addr["is_primary"])
			})

		box_dict = box.model_dump()
		box_dict["location_info"] = location_info
		return box_dict
	except Exception as e:
		logger.exception("查询电箱地址信息失败，box_id=%s", box_id)
		# 如果地址查询失败，仍返回电箱基本信息
		box_dict = box.model_dump()
		box_dict["location_info"] = []
		return box_dict


async def get_box_list(db: DatabaseService, query: DeviceQuery) -> Tuple[int, List[Box]]:
	where = ["deleted = 0"]
	params: List[Any] = []

	if query.keyword:
		where.append("(code LIKE %s OR serial_no LIKE %s)")
		params.extend([f"%{query.keyword}%", f"%{query.keyword}%"])
	if query.status:
		where.append("status = %s")
		params.append(query.status.value)
	if query.manager_code is not None:
		where.append("manager_code = %s")
		params.append(query.manager_code)
	if query.station_id:
		where.append("station_id = %s")
		params.append(query.station_id)

	# 处理区域过滤
	region_prefix = get_region_prefix(query)
	if region_prefix:
		where.append(
			"id IN (SELECT device_id FROM dev_device_address WHERE region_code LIKE %s AND device_type = 'BOX')")
		params.append(f"{region_prefix}%")

	where_clause = " AND ".join(where)
	count_sql = f"SELECT COUNT(*) AS count FROM dev_box WHERE {where_clause}"

	try:
		count_row = db.get_one(count_sql, params)
		total = count_row["count"] if count_row else 0
		if total == 0:
			return 0, []

		offset = (query.page - 1) * query.size
		select_sql = f"""
            SELECT id, station_id, code, serial_no, install_date, status, manager_code,
                   address_id, created_at, updated_at
            FROM dev_box
            WHERE {where_clause}
            ORDER BY created_at DESC
            LIMIT %s OFFSET %s
        """
		params.extend([query.size, offset])
		rows = db.get_all(select_sql, params)

		boxes = []
		for row in rows:
			row["id"] = int(row["id"])
			row["station_id"] = int(row["station_id"])
			boxes.append(Box(**row))
		return total, boxes
	except Exception as e:
		logger.exception("分页查询电箱失败，query=%s", query)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电箱列表失败") from e


async def create_box(db: DatabaseService, payload: BoxCreate) -> int:
	"""
	新建电箱接口：
	入参为基本信息、台区id、base_region_address的id
	后端新增设备记录和dev_device_address记录(AddressBindingService处理)
	"""
	if await get_box_by_code(db, payload.code):
		logger.warning("创建电箱失败，编码已存在：%s", payload.code)
		raise BizException.from_error_code(ErrorCode.CONFLICT, message=f"电箱编码 {payload.code} 已存在")

	try:
		async with db.transaction():

			sql = """
				INSERT INTO dev_box (station_id, code, serial_no, install_date, status, manager_code, address_id)
				VALUES (%s, %s, %s, %s, %s, %s, %s)
			"""

			box_id = db.insert(sql, (
				payload.station_id,
				payload.code,
				payload.serial_no,
				payload.install_date,
				payload.status.value if payload.status else DeviceStatus.ONLINE.value,
				payload.manager_code,
				payload.address_id
			))

			# 绑定地址
			await AddressService.bind(db, DeviceType.BOX.value, box_id, payload.serial_no, payload.address_id)

			logger.info("电箱创建成功，box_id=%s, address_id=%s", box_id, payload.address_id)
			return box_id

	except Exception as e:
		logger.exception("创建电箱失败，payload=%s", payload)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="创建电箱失败") from e


async def update_box(db: DatabaseService, box_id: int, payload: BoxUpdate) -> bool:
	"""
	更新电箱接口：入参为基本信息、上级设备信息，地址信息统一走 AddressBindingService.change_address
	"""
	box = await get_box_by_id(db, box_id)
	if not box:
		return False

	fields, params = [], []
	if payload.station_id is not None:
		fields.append("station_id = %s")
		params.append(payload.station_id)
	if payload.install_date is not None:
		fields.append("install_date = %s")
		params.append(payload.install_date)
	if payload.status is not None:
		fields.append("status = %s")
		params.append(payload.status.value)
	if payload.manager_code is not None:
		fields.append("manager_code = %s")
		params.append(payload.manager_code)
	if payload.serial_no is not None:
		fields.append("serial_no = %s")
		params.append(payload.serial_no)

	# 处理地址变更
	if hasattr(payload, 'address_id') and payload.address_id is not None and payload.address_id != box.address_id:
		await AddressService.change_address(db, DeviceType.BOX.value, box_id, box.serial_no, payload.address_id)
		fields.append("address_id = %s")
		params.append(payload.address_id)
		logger.info("电箱地址变更，box_id=%s, old_address_id=%s, new_address_id=%s",
					box_id, box.address_id, payload.address_id)

	if not fields:
		return True
	params.append(box_id)

	sql = f"UPDATE dev_box SET {', '.join(fields)}, updated_at = CURRENT_TIMESTAMP WHERE id = %s AND deleted = 0"
	try:
		return db.update(sql, params) > 0
	except Exception as e:
		logger.exception("更新电箱失败，box_id=%s, payload=%s", box_id, payload)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="更新电箱失败") from e


async def delete_box(db: DatabaseService, box_id: int) -> bool:
	"""
	逻辑删除电箱，同时物理删除相关的设备地址记录
	"""
	if not await get_box_by_id(db, box_id):
		return False

	try:
		async with db.transaction():
			# 逻辑删除相关的设备地址记录
			delete_address_sql = "UPDATE dev_device_address SET deleted = 1, ended_at = CURRENT_TIMESTAMP WHERE device_type = 'BOX' AND device_id = %s"
			deleted_count = db.update(delete_address_sql, (box_id,))
			logger.info("已物理删除电箱 %s 的设备地址记录，删除数量: %s", box_id, deleted_count)

			# 逻辑删除电箱
			sql = "UPDATE dev_box SET deleted = 1, updated_at = CURRENT_TIMESTAMP WHERE id = %s"
			result = db.update(sql, (box_id,))

			logger.info("成功删除电箱，box_id=%s", box_id)
			return result > 0
	except Exception as e:
		logger.exception("删除电箱失败，box_id=%s", box_id)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="删除电箱失败") from e


# ---------- RTU 操作 ----------

async def get_rtu_by_code(db: DatabaseService, code: str) -> Optional[RTU]:
	"""根据 RTU 编码（中间段）获取 RTU"""
	sql = """
        SELECT id, box_id, code, serial_no, model, station_id, address_id, comm_protocol,
               uplink_type, uplink_no, status, manager_code, last_online, created_at, updated_at
        FROM dev_rtu
        WHERE code = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (code,))
		if not row:
			return None
		row["id"] = int(row["id"])
		row["box_id"] = int(row["box_id"])
		return RTU(**row)
	except Exception as e:
		logger.exception("根据编码查询RTU失败，code=%s", code)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询RTU失败") from e


async def get_rtu_by_id(db: DatabaseService, rtu_id: int) -> Optional[RTU]:
	sql = """
        SELECT r.id, r.box_id, r.code, r.serial_no, r.model, r.collector_type, r.station_id, r.address_id, r.comm_protocol,
               r.uplink_type, r.uplink_no, r.status, r.manager_code, r.last_online, r.created_at, r.updated_at,
               b.serial_no AS box_no, s.name AS station_name
        FROM dev_rtu r
        INNER JOIN dev_box b ON b.id = r.box_id
        INNER JOIN dev_station s ON s.id = b.station_id
        WHERE r.id = %s AND r.deleted = 0
    """
	try:
		row = db.get_one(sql, (rtu_id,))
		if not row:
			return None
		return RTU(**row)
	except Exception as e:
		logger.exception("查询RTU失败，rtu_id=%s", rtu_id)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询RTU失败") from e


async def get_rtu_by_serial_no(db: DatabaseService, serial_no: str) -> Optional[RTU]:
	sql = """
        SELECT id, box_id, code, serial_no, model, station_id, address_id, comm_protocol,
               uplink_type, uplink_no, status, manager_code, last_online, created_at, updated_at
        FROM dev_rtu
        WHERE serial_no = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (serial_no,))
		if not row:
			return None
		row["id"] = int(row["id"])
		row["box_id"] = int(row["box_id"])
		return RTU(**row)
	except Exception as e:
		logger.exception("查询RTU失败，serial_no=%s", serial_no)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询RTU失败") from e


async def get_rtu_with_address(db: DatabaseService, rtu_id: int) -> Optional[Dict[str, Any]]:
	"""获取RTU及其地址信息，确保保留station_name字段"""
	rtu = await get_rtu_by_id(db, rtu_id)
	if not rtu:
		return None

	# 获取地址信息
	address_sql = """
        SELECT da.address_type, da.is_primary, da.description,
               ba.id as address_id, ba.name as address_name, ba.full_name,
               ba.longitude, ba.latitude
        FROM dev_device_address da
        INNER JOIN base_region_address ba ON da.address_id = ba.id
        WHERE da.device_type = 'RTU' AND da.device_id = %s AND da.deleted = 0 AND ba.is_active = 1
        ORDER BY da.is_primary DESC, da.created_at ASC
    """
	try:
		address_results = db.get_all(address_sql, (rtu_id,))
		location_info = []
		for addr in address_results:
			location_info.append({
				"address_id": addr["address_id"],
				"address_name": addr["address_name"],
				"full_address": addr["full_name"],
				"longitude": addr["longitude"],
				"latitude": addr["latitude"],
				"address_type": addr["address_type"],
				"is_primary": bool(addr["is_primary"])
			})

		rtu_dict = dict(rtu)
		if hasattr(rtu, 'station_name'):
			rtu_dict['station_name'] = rtu.station_name
		rtu_dict["location_info"] = location_info
		return rtu_dict
	except BizException:
		logger.exception("查询RTU地址信息失败，rtu_id=%s", rtu_id)
		# 如果地址查询失败，仍返回RTU基本信息
		try:
			rtu_dict = dict(rtu)
			if hasattr(rtu, 'station_name'):
				rtu_dict['station_name'] = rtu.station_name
			rtu_dict["location_info"] = []
			return rtu_dict
		except BizException:
			pass
		return None


async def get_rtu_list(db: DatabaseService, query: DeviceQuery) -> Tuple[int, List[RTU]]:
	where = ["r.deleted = 0"]
	params: List[Any] = []

	if query.keyword:
		where.append("(r.serial_no LIKE %s OR r.model LIKE %s)")
		params.extend([f"%{query.keyword}%", f"%{query.keyword}%"])
	if query.status:
		where.append("r.status = %s")
		params.append(query.status.value)
	if query.manager_code is not None:
		where.append("r.manager_code = %s")
		params.append(query.manager_code)
	if query.box_id:
		where.append("r.box_id = %s")
		params.append(query.box_id)

	# 处理区域过滤
	region_prefix = get_region_prefix(query)
	if region_prefix:
		where.append(
			"r.id IN (SELECT device_id FROM dev_device_address WHERE region_code LIKE %s AND device_type = 'RTU')")
		params.append(f"{region_prefix}%")

	# 处理台区过滤
	if query.station_id:
		where.append(
			"r.box_id IN (SELECT id FROM dev_box WHERE station_id = %s AND deleted = 0)")
		params.append(query.station_id)

	where_clause = " AND ".join(where)
	count_sql = f"SELECT COUNT(*) AS count FROM dev_rtu r WHERE {where_clause}"

	try:
		count_row = db.get_one(count_sql, params)
		total = count_row["count"] if count_row else 0
		if total == 0:
			return 0, []

		offset = (query.page - 1) * query.size
		select_sql = f"""
            SELECT r.id, r.box_id, r.code, r.serial_no, r.model, r.collector_type, r.station_id, r.comm_protocol, r.uplink_type, r.uplink_no,
                   r.manager_code, r.last_online, r.address_id, r.status, r.created_at, r.updated_at,
                   b.serial_no AS box_no, s.name AS station_name
			FROM dev_rtu r
			INNER JOIN dev_box b ON b.id = r.box_id
			INNER JOIN dev_station s ON s.id = b.station_id
            WHERE {where_clause}
            ORDER BY r.created_at DESC
            LIMIT %s OFFSET %s
        """
		params.extend([query.size, offset])
		rows = db.get_all(select_sql, params)

		rtus = []
		for row in rows:
			rtus.append(RTU(**row))
		return total, rtus
	except Exception as e:
		logger.exception("分页查询RTU失败，query=%s", query)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询RTU列表失败") from e


async def create_rtu(db: DatabaseService, payload: RTUCreate) -> int:
	"""
	新建RTU接口：
	入参为基本信息、台区id、base_region_address的id
	后端新增设备记录和dev_device_address记录(AddressBindingService处理)
	"""
	if await get_rtu_by_code(db, payload.code):
		logger.warning("创建RTU失败，编码已存在：%s", payload.code)
		raise BizException.from_error_code(ErrorCode.CONFLICT, message=f"RTU编码 {payload.code} 已存在")
	if payload.serial_no and await get_rtu_by_serial_no(db, payload.serial_no):
		logger.warning("创建RTU失败，序列号已存在：%s", payload.serial_no)
		raise BizException.from_error_code(ErrorCode.CONFLICT, message=f"RTU序列号 {payload.serial_no} 已存在")

	sql = """
        INSERT INTO dev_rtu (
            box_id, code, serial_no, model, collector_type, station_id, address_id, comm_protocol,
            uplink_type, uplink_no, status, last_online, manager_code
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """
	try:
		rtu_id = db.insert(sql, (
			payload.box_id,
			payload.code,
			payload.serial_no,
			payload.model,
			payload.collector_type,
			payload.station_id,
			payload.address_id,
			payload.comm_protocol,
			payload.uplink_type.value if payload.uplink_type else None,
			payload.uplink_no,
			payload.status.value if payload.status else DeviceStatus.ONLINE.value,
			datetime.now(),
			payload.manager_code
		))

		# 绑定地址
		await AddressService.bind(db, DeviceType.RTU.value, rtu_id, payload.serial_no, payload.address_id)

		logger.info("RTU创建成功，rtu_id=%s, address_id=%s", rtu_id, payload.address_id)
		return rtu_id
	except Exception as e:
		logger.exception("创建RTU失败，payload=%s", payload)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="创建RTU失败") from e


async def update_rtu(db: DatabaseService, rtu_id: int, payload: RTUUpdate) -> bool:
	"""
	更新RTU接口：入参为基本信息、上级设备信息，地址信息统一走 AddressBindingService.change_address
	"""
	rtu = await get_rtu_by_id(db, rtu_id)
	if not rtu:
		return False

	fields, params = [], []
	if payload.box_id is not None:
		fields.append("box_id = %s")
		params.append(payload.box_id)
	if payload.model is not None:
		fields.append("model = %s")
		params.append(payload.model)
	if payload.collector_type is not None:
		fields.append("collector_type = %s")
		params.append(payload.collector_type)
	if payload.station_id is not None:
		fields.append("station_id = %s")
		params.append(payload.station_id)
	if payload.comm_protocol is not None:
		fields.append("comm_protocol = %s")
		params.append(payload.comm_protocol)
	if payload.uplink_type is not None:
		fields.append("uplink_type = %s")
		params.append(payload.uplink_type)
	if payload.uplink_no is not None:
		fields.append("uplink_no = %s")
		params.append(payload.uplink_no)
	if payload.status is not None:
		fields.append("status = %s")
		params.append(payload.status.value)
	if payload.manager_code is not None:
		fields.append("manager_code = %s")
		params.append(payload.manager_code)
	if payload.serial_no is not None:
		fields.append("serial_no = %s")
		params.append(payload.serial_no)

	# 处理地址变更
	if hasattr(payload, 'address_id') and payload.address_id is not None and payload.address_id != rtu.address_id:
		await AddressService.change_address(db, DeviceType.RTU.value, rtu_id, rtu.serial_no, payload.address_id)
		fields.append("address_id = %s")
		params.append(payload.address_id)
		logger.info("RTU地址变更，rtu_id=%s, old_address_id=%s, new_address_id=%s",
					rtu_id, rtu.address_id, payload.address_id)

	if not fields:
		return True
	params.append(rtu_id)

	sql = f"UPDATE dev_rtu SET {', '.join(fields)}, updated_at = CURRENT_TIMESTAMP WHERE id = %s AND deleted = 0"
	try:
		return db.update(sql, params) > 0
	except Exception as e:
		logger.exception("更新RTU失败，rtu_id=%s, payload=%s", rtu_id, payload)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="更新RTU失败") from e


async def delete_rtu(db: DatabaseService, rtu_id: int) -> bool:
	"""
	逻辑删除RTU，同时物理删除相关的设备地址记录
	"""
	if not await get_rtu_by_id(db, rtu_id):
		return False

	try:
		async with db.transaction():
			# 物理删除相关的设备地址记录
			delete_address_sql = "UPDATE dev_device_address SET deleted = 1, ended_at = CURRENT_TIMESTAMP WHERE device_id = %s AND device_type = 'RTU' "
			deleted_count = db.update(delete_address_sql, (rtu_id,))
			logger.info("已物理删除RTU %s 的设备地址记录，删除数量: %s", rtu_id, deleted_count)

			# 逻辑删除RTU
			sql = "UPDATE dev_rtu SET deleted = 1, updated_at = CURRENT_TIMESTAMP WHERE id = %s"
			result = db.update(sql, (rtu_id,))

			logger.info("成功删除RTU，rtu_id=%s", rtu_id)
			return result > 0
	except Exception as e:
		logger.exception("删除RTU失败，rtu_id=%s", rtu_id)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="删除RTU失败") from e


# ---------- 电表操作 ----------

async def get_meter_by_id(db: DatabaseService, meter_id: int) -> Optional[Meter]:
	sql = """
        SELECT m.id, m.rtu_id, m.code, m.serial_no, m.meter_type, m.phase, m.station_id, m.address_id,
               m.ct_ratio, m.pt_ratio, m.init_reading, m.status,
               m.commissioning_date, m.calibration_date, m.calibration_due_date,
               m.created_at, m.updated_at, r.serial_no AS rtu_rtu_no, r.box_id
        FROM dev_meter m
        LEFT JOIN dev_rtu r ON r.id = m.rtu_id
        WHERE m.id = %s AND m.deleted = 0
    """
	try:
		row = db.get_one(sql, (meter_id,))
		if not row:
			return None

		# 补充 box 和 station 信息
		box = await get_box_by_id(db, row["box_id"])
		if box:
			row["box_serial_no"] = box.serial_no
			row["box_id"] = box.id

			station = await get_station_by_id(db, box.station_id)
			if station:
				row["station_serial_no"] = station.serial_no
				row["station_id"] = station.id
		else:
			row["box_serial_no"] = None
			row["station_serial_no"] = None

		return Meter(**row)
	except Exception as e:
		logger.exception("查询电表失败，meter_id=%s", meter_id)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电表失败") from e


async def get_meter_by_code(db: DatabaseService, code: str) -> Optional[Meter]:
	sql = """
        SELECT id, rtu_id, code, serial_no, meter_type, phase, station_id, address_id,
               ct_ratio, pt_ratio, init_reading, status,
               commissioning_date, calibration_date, calibration_due_date, created_at, updated_at
        FROM dev_meter
        WHERE code = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (code,))
		if not row:
			return None
		row["id"] = int(row["id"])
		row["rtu_id"] = int(row["rtu_id"])
		return Meter(**row)
	except Exception as e:
		logger.exception("查询电表失败，code=%s", code)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电表失败") from e


async def get_meter_by_serial_no(db: DatabaseService, serial_no: str) -> Optional[Meter]:
	sql = """
        SELECT id, rtu_id, code, serial_no, meter_type, phase, station_id, address_id,
               ct_ratio, pt_ratio, init_reading, status,
               commissioning_date, calibration_date, calibration_due_date, created_at, updated_at
        FROM dev_meter
        WHERE serial_no = %s AND deleted = 0
    """
	try:
		row = db.get_one(sql, (serial_no,))
		if not row:
			return None
		row["id"] = int(row["id"])
		row["rtu_id"] = int(row["rtu_id"])
		return Meter(**row)
	except Exception as e:
		logger.exception("查询电表失败，serial_no=%s", serial_no)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电表失败") from e


async def get_meter_with_address(db: DatabaseService, meter_id: int) -> Optional[Dict[str, Any]]:
	"""获取电表及其地址信息"""
	meter = await get_meter_by_id(db, meter_id)
	if not meter:
		return None

	# 获取地址信息
	address_sql = """
        SELECT da.address_type, da.is_primary, da.description,
               ba.id as address_id, ba.name as address_name, ba.full_name,
               ba.longitude, ba.latitude
        FROM dev_device_address da
        LEFT JOIN base_region_address ba ON da.address_id = ba.id
        WHERE da.device_type = 'METER' AND da.device_id = %s AND da.deleted = 0
        ORDER BY da.is_primary DESC, da.created_at ASC
    """
	try:
		address_results = db.get_all(address_sql, (meter_id,))
		location_info = []
		for addr in address_results:
			location_info.append({
				"address_id": addr["address_id"],
				"address_name": addr["address_name"],
				"full_address": addr["full_name"],
				"longitude": addr["longitude"],
				"latitude": addr["latitude"],
				"address_type": addr["address_type"],
				"is_primary": bool(addr["is_primary"])
			})

		meter_dict = meter.model_dump()
		meter_dict["location_info"] = location_info
		return meter_dict
	except Exception as e:
		logger.exception("查询电表地址信息失败，meter_id=%s", meter_id)
		# 如果地址查询失败，仍返回电表基本信息
		meter_dict = meter.model_dump()
		meter_dict["location_info"] = []
		return meter_dict


async def get_meter_list(db: DatabaseService, query: DeviceQuery) -> Tuple[int, List[Meter]]:
	where = ["m.deleted = 0"]
	params: List[Any] = []

	if query.keyword:
		where.append("(m.code LIKE %s OR m.serial_no LIKE %s)")
		params.extend([f"%{query.keyword}%", f"%{query.keyword}%"])
	if query.status:
		where.append("m.status = %s")
		params.append(query.status.value)
	if query.station_id:
		where.append("m.station_id = %s")
		params.append(query.station_id)
	if query.rtu_id:
		where.append("m.rtu_id = %s")
		params.append(query.rtu_id)
	if query.phase:
		where.append("m.phase = %s")
		params.append(query.phase.value)

	# 处理区域过滤
	region_prefix = get_region_prefix(query)
	if region_prefix:
		where.append(
			"m.id IN (SELECT device_id FROM dev_device_address WHERE region_code LIKE %s AND device_type = 'METER')")
		params.append(f"{region_prefix}%")

	where_clause = " AND ".join(where)
	count_sql = f"SELECT COUNT(*) AS count FROM dev_meter m WHERE {where_clause}"

	try:
		count_row = db.get_one(count_sql, params)
		total = count_row["count"] if count_row else 0
		if total == 0:
			return 0, []

		offset = (query.page - 1) * query.size
		select_sql = f"""
            SELECT m.id, m.code, m.serial_no, m.rtu_id, r.serial_no as rtu_serial_no, m.meter_type, m.phase,
                   m.station_id, m.address_id, m.ct_ratio, m.pt_ratio, m.init_reading, m.status,
                   m.commissioning_date, m.calibration_date, m.calibration_due_date, m.created_at, m.updated_at,
                   b.serial_no As box_serial_no, b.id AS box_id
            FROM dev_meter m
            INNER JOIN dev_rtu r ON r.id = m.rtu_id
            INNER JOIN dev_box b ON r.box_id = b.id
            WHERE {where_clause}
            ORDER BY m.created_at DESC
            LIMIT %s OFFSET %s
        """
		params.extend([query.size, offset])
		rows = db.get_all(select_sql, params)

		meters = []
		for row in rows:
			meters.append(Meter(**row))
		return total, meters
	except Exception as e:
		logger.exception("分页查询电表失败，query=%s", query)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电表列表失败") from e


async def create_meter(db: DatabaseService, payload: MeterCreate) -> int:
	"""
	新建电表接口：
	入参为基本信息、台区id、base_region_address的id
	后端新增设备记录和dev_device_address记录(AddressBindingService处理)
	如果是单相表,还需保存到dev_meter_hierarchy表
	"""
	if await get_meter_by_code(db, payload.code):
		logger.warning("创建电表失败，编码已存在：%s", payload.code)
		raise BizException.from_error_code(ErrorCode.CONFLICT, message=f"电表编码 {payload.code} 已存在")
	if payload.serial_no and await get_meter_by_serial_no(db, payload.serial_no):
		logger.warning("创建电表失败，序列号已存在：%s", payload.serial_no)
		raise BizException.from_error_code(ErrorCode.CONFLICT, message=f"电表序列号 {payload.serial_no} 已存在")

	sql = """
        INSERT INTO dev_meter (
            rtu_id, code, serial_no, meter_type, phase, station_id, box_id, address_id, ct_ratio, pt_ratio,
            init_reading, status, manager_code, commissioning_date, calibration_date, calibration_due_date
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """
	try:
		meter_id = db.insert(sql, (
			payload.rtu_id,
			payload.code,
			payload.serial_no,
			payload.meter_type.value if payload.meter_type else None,
			payload.phase.value if payload.phase else None,
			payload.station_id,
			payload.box_id,
			payload.address_id,
			payload.ct_ratio,
			payload.pt_ratio,
			payload.init_reading,
			payload.status.value if payload.status else DeviceStatus.ONLINE.value,
			payload.manager_code,
			payload.commissioning_date,
			payload.calibration_date,
			payload.calibration_due_date
		))

		# 绑定地址
		await AddressService.bind(db, DeviceType.METER.value, meter_id, payload.serial_no, payload.address_id)

		# 如果是单相表且指定了上级三相表,保存层级关系
		if payload.meter_type and payload.meter_type.value == 'SINGLE' and payload.parent_meter_id:
			# 查询三相表的RTU信息
			parent_meter_sql = """
				SELECT m.id, r.serial_no as rtu_serial_no
				FROM dev_meter m
				INNER JOIN dev_rtu r ON m.rtu_id = r.id
				WHERE m.id = %s AND m.deleted = 0
			"""
			parent_row = db.get_one(parent_meter_sql, (payload.parent_meter_id,))
			if not parent_row:
				raise BizException.from_error_code(ErrorCode.NOT_FOUND,
												   message=f"上级三相表 {payload.parent_meter_id} 不存在")

			# 插入层级关系
			hierarchy_sql = """
				INSERT INTO dev_meter_hierarchy (rtu_serial_no, parent_meter_id, child_meter_id, phase)
				VALUES (%s, %s, %s, %s)
			"""
			db.insert(hierarchy_sql, (
				parent_row['rtu_serial_no'],
				payload.parent_meter_id,
				meter_id,
				payload.phase.value
			))

		logger.info("电表创建成功，meter_id=%s, box_id=%s, address_id=%s", meter_id, payload.box_id,
					payload.address_id)
		return meter_id
	except Exception as e:
		logger.exception("创建电表失败，payload=%s", payload)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="创建电表失败") from e


async def update_meter(db: DatabaseService, meter_id: int, payload: MeterUpdate) -> bool:
	"""
	更新电表接口：
	1. 支持更新基本信息
	2. 支持根据station_id、box_id、rtu_id更新电表所属关系
	3. 地址信息统一走 AddressBindingService.change_address
	"""
	meter = await get_meter_by_id(db, meter_id)
	if not meter:
		return False

	fields, params = [], []

	# 处理上级设备关系变更
	if payload.station_id is not None or payload.box_id is not None or payload.rtu_id is not None:

		final_rtu_id = None
		final_box_id = None
		final_station_id = None

		if payload.rtu_id is not None:
			# 直接使用传入的 rtu_id
			final_rtu_id = payload.rtu_id
			rtu = await get_rtu_by_id(db, final_rtu_id)
			if not rtu:
				raise BizException.from_error_code(ErrorCode.NOT_FOUND, message=f"RTU ID {final_rtu_id} 不存在")
			final_box_id = rtu.box_id

		if payload.box_id is not None:
			box = await get_box_by_id(db, payload.box_id)
			if not box:
				raise BizException.from_error_code(ErrorCode.NOT_FOUND, message=f"电箱 ID {payload.box_id} 不存在")

			# 查找电箱下是否存在RTU
			statuon_sql = """
				SELECT r.id, r.station_id
				FROM dev_rtu r
				INNER JOIN dev_box b ON r.box_id = b.id
				WHERE b.id = %s
				  AND r.id = %s	
				  AND r.deleted = 0
				  AND r.status != %s
				  AND b.deleted = 0
				LIMIT 1
			"""
			station_row = db.get_one(statuon_sql, (payload.box_id, payload.rtu_id, DeviceStatus.DECOMMISSIONED.value))
			if not station_row:
				raise BizException.from_error_code(
					ErrorCode.NOT_FOUND,
					message=f"电箱下未找到可用的 RTU"
				)

			final_box_id = box.id

		if payload.station_id is not None:
			station = await get_station_by_id(db, payload.station_id)
			if not station:
				raise BizException.from_error_code(ErrorCode.NOT_FOUND, message=f"台区 ID {payload.station_id} 不存在")

			statuon_sql = """
				SELECT s.id
				FROM dev_station s
				INNER JOIN dev_box b ON s.id = b.station_id
				WHERE b.station_id = %s
				  AND b.id = %s
				  AND b.deleted = 0
				  AND b.status != %s
				  AND s.deleted = 0
				LIMIT 1
			"""
			station_row = db.get_one(statuon_sql,
									 (payload.station_id, payload.box_id, DeviceStatus.DECOMMISSIONED.value))
			if not station_row:
				raise BizException.from_error_code(
					ErrorCode.NOT_FOUND,
					message=f"台区 下未找到可用的 BOX"
				)

			final_station_id = payload.station_id

		if final_station_id is not None and final_station_id != meter.station_id:
			fields.append("station_id = %s")
			params.append(final_station_id)

		if final_box_id is not None and final_station_id != meter.box_id:
			fields.append("box_id = %s")
			params.append(final_box_id)

		if final_rtu_id is not None and final_rtu_id != meter.rtu_id:
			fields.append("rtu_id = %s")
			params.append(final_rtu_id)

	# 处理其他字段更新
	for field, value in payload.model_dump(exclude_unset=True).items():
		# 跳过已处理的上级设备字段和层级关系字段
		if field in ['station_id', 'box_id', 'rtu_id', 'parent_meter_id']:
			continue

		if field == "status":
			fields.append("status = %s")
			params.append(value.value)
		elif field == "address_id":
			# 处理地址变更
			if value != meter.address_id:
				await AddressService.change_address(db, DeviceType.METER.value, meter_id, meter.serial_no, value)
				fields.append("address_id = %s")
				params.append(value)
				logger.info("电表地址变更，meter_id=%s, old_address_id=%s, new_address_id=%s",
							meter_id, meter.address_id, value)
		# 处理列表类型的字段（文件相关）
		elif field in ['file_ids', 'add_file_ids', 'del_file_ids', 'parent_device_info'] and value is not None:
			# 这些字段在其他地方处理，这里跳过
			continue
		elif field == "serial_no":
			fields.append("serial_no = %s")
			params.append(value)
		elif field == "meter_type":
			if isinstance(value, str):
				fields.append("meter_type = %s")
				params.append(value)
			else:
				fields.append("meter_type = %s")
				params.append(value.value if hasattr(value, 'value') else value)
		elif field == "phase":
			if isinstance(value, str):
				fields.append("phase = %s")
				params.append(value)
			else:
				fields.append("phase = %s")
				params.append(value.value if hasattr(value, 'value') else value)
		else:
			# 其他常规字段
			fields.append(f"{field} = %s")
			params.append(value)

	if not fields:
		return True

	# 处理层级关系更新(单相表)
	if payload.parent_meter_id is not None:
		# 先删除旧的层级关系
		delete_hierarchy_sql = "DELETE FROM dev_meter_hierarchy WHERE child_meter_id = %s"
		db.update(delete_hierarchy_sql, (meter_id,))

		# 如果parent_meter_id>0,插入新的层级关系
		if payload.parent_meter_id > 0:
			# 查询三相表的RTU信息
			parent_meter_sql = """
				SELECT m.id, r.serial_no as rtu_serial_no
				FROM dev_meter m
				INNER JOIN dev_rtu r ON m.rtu_id = r.id
				WHERE m.id = %s AND m.deleted = 0
			"""
			parent_row = db.get_one(parent_meter_sql, (payload.parent_meter_id,))
			if not parent_row:
				raise BizException.from_error_code(ErrorCode.NOT_FOUND,
												   message=f"上级三相表 {payload.parent_meter_id} 不存在")

			# 插入新的层级关系
			hierarchy_sql = """
				INSERT INTO dev_meter_hierarchy (rtu_serial_no, parent_meter_id, child_meter_id, phase)
				VALUES (%s, %s, %s, %s)
			"""
			db.insert(hierarchy_sql, (
				parent_row['rtu_serial_no'],
				payload.parent_meter_id,
				meter_id,
				payload.phase
			))
		else:
			logger.info(f"删除电表层级关系: child_meter_id={meter_id}")

	params.append(meter_id)

	sql = f"UPDATE dev_meter SET {', '.join(fields)}, updated_at = CURRENT_TIMESTAMP WHERE id = %s AND deleted = 0"
	try:
		result = db.update(sql, params)
		logger.info("电表更新成功，meter_id=%s, 更新字段数=%s", meter_id, len(fields))
		return result > 0
	except Exception as e:
		logger.exception("更新电表失败，meter_id=%s, payload=%s", meter_id, payload)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="更新电表失败") from e


async def delete_meter(db: DatabaseService, meter_id: int) -> bool:
	if not await get_meter_by_id(db, meter_id):
		return False

	try:
		async with db.transaction():
			# 删除相关的设备地址记录
			delete_address_sql = "UPDATE dev_device_address SET deleted = 1, ended_at = CURRENT_TIMESTAMP WHERE device_id = %s AND device_type = 'METER' "
			deleted_count = db.update(delete_address_sql, (meter_id,))
			logger.info("已逻辑删除电表 %s 的设备地址记录，删除数量: %s", meter_id, deleted_count)

			sql = "UPDATE dev_meter SET deleted = 1, updated_at = CURRENT_TIMESTAMP WHERE id = %s"

			return (db.update(sql, (meter_id,))) > 0
	except BizException as e:
		logger.exception("删除电表失败，meter_id=%s", meter_id)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="删除电表失败") from e


# ---------- 设备负责人操作 ----------

async def update_station_manager(db: DatabaseService, station_id: int, manager_code: str) -> bool:
	"""更新台区负责人"""
	station = await get_station_by_id(db, station_id)
	if not station:
		return False

	sql = "UPDATE dev_station SET manager_code = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s AND deleted = 0"
	try:
		return db.update(sql, (manager_code, station_id)) > 0
	except Exception as e:
		logger.exception("更新台区负责人失败，station_id=%s, manager_code=%s", station_id, manager_code)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="更新台区负责人失败") from e


async def update_box_manager(db: DatabaseService, box_id: int, manager_code: str) -> bool:
	"""更新电箱负责人"""
	box = await get_box_by_id(db, box_id)
	if not box:
		return False

	sql = "UPDATE dev_box SET manager_code = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s AND deleted = 0"
	try:
		return db.update(sql, (manager_code, box_id)) > 0
	except Exception as e:
		logger.exception("更新电箱负责人失败，box_id=%s, manager_code=%s", box_id, manager_code)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="更新电箱负责人失败") from e


async def update_rtu_manager(db: DatabaseService, rtu_id: int, manager_code: str) -> bool:
	"""更新RTU负责人"""
	rtu = await get_rtu_by_id(db, rtu_id)
	if not rtu:
		return False

	sql = "UPDATE dev_rtu SET manager_code = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s AND deleted = 0"
	try:
		return db.update(sql, (manager_code, rtu_id)) > 0
	except Exception as e:
		logger.exception("更新RTU负责人失败，rtu_id=%s, manager_code=%s", rtu_id, manager_code)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="更新RTU负责人失败") from e


# ---------- RTU替换与报废SQL操作 ----------

async def migrate_meters_to_new_rtu(db: DatabaseService, old_rtu_id: int, new_rtu_id: int, new_rtu_serial_no: str) -> int:
	"""将旧RTU下的电表迁移到新RTU
	
	Args:
		db: 数据库服务
		old_rtu_id: 旧RTU ID
		new_rtu_id: 新RTU ID
		new_rtu_serial_no: 新RTU序列号
	
	Returns:
		int: 迁移的电表数量
	"""
	try:
		# 查询电表数量
		count_sql = "SELECT COUNT(*) as count FROM dev_meter WHERE rtu_id = %s AND deleted = 0"
		count_row = db.get_one(count_sql, (old_rtu_id,))
		affected_count = count_row["count"] if count_row else 0
		
		if affected_count > 0:
			# 批量更新电表的rtu_id
			update_sql = """
				UPDATE dev_meter
				SET rtu_id = %s, updated_at = NOW()
				WHERE rtu_id = %s AND deleted = 0
			"""
			db.update(update_sql, (new_rtu_id, old_rtu_id))
			
			# 更新dev_meter_hierarchy表中的rtu_serial_no
			# 需要更新两种情况:
			# 1. parent_meter_id对应的三相表属于该RTU
			# 2. child_meter_id对应的单相表属于该RTU
			update_hierarchy_sql = """
				UPDATE dev_meter_hierarchy h
				INNER JOIN dev_meter m ON (h.parent_meter_id = m.id OR h.child_meter_id = m.id)
				SET h.rtu_serial_no = %s
				WHERE m.rtu_id = %s AND m.deleted = 0
			"""
			db.update(update_hierarchy_sql, (new_rtu_serial_no, new_rtu_id))
			logger.info(f"RTU替换: 将{affected_count}个电表从旧RTU({old_rtu_id})迁移到新RTU({new_rtu_id}),并更新层级关系表")
		
		return affected_count
	except Exception as e:
		logger.exception("迁移电表到新RTU失败，old_rtu_id=%s, new_rtu_id=%s", old_rtu_id, new_rtu_id)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="迁移电表失败") from e


async def get_meters_by_rtu_id(db: DatabaseService, rtu_id: int) -> List[dict]:
	"""根据RTU ID获取绑定的电表列表
	
	Returns:
		List[dict]: 电表列表, 每个元素包含 id 和 serial_no
	"""
	sql = "SELECT id, serial_no FROM dev_meter WHERE rtu_id = %s AND deleted = 0"
	try:
		rows = db.get_all(sql, (rtu_id,))
		return rows if rows else []
	except Exception as e:
		logger.exception("查询RTU绑定的电表失败，rtu_id=%s", rtu_id)
		raise BizException.from_error_code(ErrorCode.DB_QUERY_ERROR, message="查询电表列表失败") from e


async def unbind_meters_from_rtu(db: DatabaseService, rtu_id: int) -> int:
	"""解绑RTU的所有电表(rtu_id置空NULL)
	
	Returns:
		int: 解绑的电表数量
	"""
	unbind_sql = """
		UPDATE dev_meter
		SET rtu_id = NULL, updated_at = NOW()
		WHERE rtu_id = %s AND deleted = 0
	"""
	try:
		unbind_count = db.update(unbind_sql, (rtu_id,))
		return unbind_count
	except Exception as e:
		logger.exception("解绑RTU的电表失败，rtu_id=%s", rtu_id)
		raise BizException.from_error_code(ErrorCode.DATABASE_ERROR, message="解绑电表失败") from e
